package com.crazyice.lee.writer;

import com.crazyice.lee.data.Student;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Properties;

/**
 * @Description mysql sink
 * @Author jiangxiaozhi
 * @Date 2018/10/15 18:31
 **/
@Slf4j
public class JdbcWriter extends RichSinkFunction<Student> {

    private Connection connection;
    private PreparedStatement preparedStatement;

    private static Properties properties;

    private static void init() {
        properties = new Properties();
        try(InputStream inputStream = JdbcWriter.class.getResourceAsStream("/jdbc.properties")) {
            properties.load(inputStream);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        init();

        super.open(parameters);
        // 加载JDBC驱动
        Class.forName(properties.getProperty("DriverClass"));
        // 获取数据库连接
        connection = DriverManager.getConnection(properties.getProperty("SinkDriverUrl"),properties.getProperty("SinkUser"),properties.getProperty("SinkPassword"));
        preparedStatement = connection.prepareStatement(properties.getProperty("SinkSql"));
        super.open(parameters);
    }
 
    @Override
    public void close() throws Exception {
        super.close();
        if(preparedStatement != null){
            preparedStatement.close();
        }
        if(connection != null){
            connection.close();
        }
        super.close();
    }
 
    @Override
    public void invoke(Student value, Context context) throws Exception {
        try {
            String name=value.getName();
            String password=value.getPassword();
            int age=value.getAge();
            boolean sex=value.isSex();
            preparedStatement.setString(1,name);
            preparedStatement.setString(2,password);
            preparedStatement.setInt(3,age);
            preparedStatement.setBoolean(4,sex);
            int count=preparedStatement.executeUpdate();
            log.info("插入{}条记录。",count);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}